#pragma once

#include "../common/Log.hpp"
#include "../common/SqliteHelper.hpp"
#include "../common/FileHelper.hpp"
#include "../common/UUIDHelper.hpp"
#include "../common/message.pb.h"
#include "MessageMapper.hpp"
#include "MessageCenter.hpp"
#include <iostream>
#include <string>
#include <list>
#include <unordered_map>
#include <mutex>
#include <memory>


namespace micromq
{
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
    public:
        MessageManager(const std::string& baseDir): _baseDir(baseDir)
        {}

        void addMessageCenter(const std::string& msgQueueName)
        {
            MessageCenter::ptr msgCenter;
            {
                std::unique_lock<std::mutex> lock(_mtx); //MessageManager中的锁只应保证_centers哈希表的线程安全
                if(_centers.find(msgQueueName) != _centers.end()) return;
                msgCenter = std::make_shared<MessageCenter>(_baseDir, msgQueueName);
                _centers[msgQueueName] = msgCenter;
            }
            msgCenter->recovery();
        }

        void removeMessageCenter(const std::string& msgQueueName)
        {
            MessageCenter::ptr msgCenter;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                if(_centers.find(msgQueueName) != _centers.end())
                {
                    msgCenter = _centers[msgQueueName];
                    _centers.erase(msgQueueName);
                }
            }
            msgCenter->clear();
        }

        void produceMessage(const std::string& msgQueueName, BasicProperties* bp, const std::string& content, bool msgQueueDurable)
        {
            MessageCenter::ptr msgCenter;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _centers.find(msgQueueName);
                if(it == _centers.end())
                {
                    logInfo("produce error, message center of %s is not exists...", msgQueueName.c_str());
                    return;
                }
                msgCenter = it->second;
            }
            msgCenter->addMessage(bp, content, msgQueueDurable);
        }

        MessagePtr consumeMessage(const std::string& msgQueueName)
        {
            MessageCenter::ptr msgCenter;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _centers.find(msgQueueName);
                if(it == _centers.end())
                {
                    logInfo("consume error, message center of %s is not exists...", msgQueueName.c_str());
                    return nullptr;
                }
                msgCenter = it->second;
            }
            return msgCenter->getMessage();
        }

        void ackMessage(const std::string& msgQueueName, const std::string& msgId)
        {
            MessageCenter::ptr msgCenter;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _centers.find(msgQueueName);
                if(it == _centers.end())
                {
                    logInfo("ack error, message center of %s is not exists...", msgQueueName.c_str());
                    return;
                }
                msgCenter = it->second;
            }
            msgCenter->removeMessage(msgId);
        }

        std::unordered_map<std::string, MessageCenter::ptr> getMessageCenters()
        {
            return _centers;
        }

        int getWaitConsumeNum(const std::string& msgQueueName)
        {
            return _centers[msgQueueName]->getWaitConsumeNum();
        }

        int getWaitAckNum(const std::string& msgQueueName)
        {
            return _centers[msgQueueName]->getWaitAckNum();
        }

        int getValidDurableNum(const std::string& msgQueueName)
        {
            return _centers[msgQueueName]->getValidDurableNum();
        }

        int getTotalDurableNum(const std::string& msgQueueName)
        {
            return _centers[msgQueueName]->getTotalDurableNum();
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            for(auto& it : _centers)
            {
                it.second->clear();
            }
        }

        ~MessageManager()
        {}
    private:
        std::mutex _mtx;
        std::string _baseDir;
        std::unordered_map<std::string, MessageCenter::ptr> _centers; // <队列名, 消息中心*>
    };
}